<?php
namespace App\Common\Cla\EChat;

use App\Common as com;
use App\Common\Cla\EChat\Server as thisServer;
use PhpAmqpLib\Exchange\AMQPExchangeType;

class Server extends com\Cla\ClaBase {

	/**
	 * @var com\Lib\Com\RabbitMQ|null
	 */
	private ?com\Lib\Com\RabbitMQ $rabbitMQ = null;

	/**
	 * @var com\Lib\Com\RabbitMQ\_Channel\_Exchange[]
	 */
	private array $exchangePool = [];

	/**
	 * 格式：
	 * @var com\Lib\Com\RabbitMQ\_Channel\_Queue[]
	 */
	private array $queuePools = [];

	private thisServer\ReceivePools $receivePools;


	/**
	 * 连接uuid
	 * @var string
	 */
	private string $serverUuid;

	public function __construct(  ) {
		$this->serverUuid = com\Lib\Fun\Str::uuid( false );
		//
		$this->receivePools = thisServer\ReceivePools::makeOfNew();
		//
		$this->initRabbitMQ();
		//
		parent::__construct();
	}

	/**
	 * @return string
	 */
	public function getServerUuid(  ):string {
		return $this->serverUuid;
	}

	private function initRabbitMQ(  ) {
		if( $this->rabbitMQ === null ){
			$this->rabbitMQ = com\Lib\Com\RabbitMQ::newOfPoolOfDefault();
		}else{
			$this->rabbitMQ->reconnect();
		}
		//
		/*
		$this->receive( 'test', 'test', function ( \PhpAmqpLib\Message\AMQPMessage $AMQPMsg ){
			$AMQPMsg->
		} );
		*/
	}

	/**
	 * 获取ChannelSid的前缀
	 * @return string
	 */
	public function getChannelSidPrefix(  ):string {
		$channelSidPrefix = config( 'eChat.channelSidPrefix' );
		if( ! empty( $channelSidPrefix ) ){
			$channelSidPrefix .= '.';
		}
		//
		return '{$channelSidPrefix}eCaht.';
	}

	/**
	 * @param string $channelSid
	 * @params array $params 参数数组
	 * ················· [可空](bool)	$params['exchangeDeclare']['passive']	说明：交换机声明。[false][默认]
	 * ················· [可空](bool)	$params['exchangeDeclare']['durable']	说明：交换机声明。[false][默认]
	 * ················· [可空](bool)	$params['exchangeDeclare']['autoDelete']	说明：交换机声明，自动删除。[true][默认]
	 * @return com\Lib\Com\RabbitMQ\_Channel\_Exchange
	 */
	public function getExchangeOfPools( string $channelSid = 'default', array $params = [] ):com\Lib\Com\RabbitMQ\_Channel\_Exchange {
		if( ! array_key_exists( $channelSid, $this->exchangePool ) ){
			$this->exchangeDeclareToPools( $channelSid );
		}
		//
		return $this->exchangePool[ $channelSid ];
	}

	/**
	 * 声明交换机（会覆盖pool）
	 * @param string $channelSid
	 * @params array $params 参数数组
	 * ················· [可空](bool)	$params['exchangeDeclare']['passive']	说明：交换机声明。[false][默认]
	 * ················· [可空](bool)	$params['exchangeDeclare']['durable']	说明：交换机声明。[false][默认]
	 * ················· [可空](bool)	$params['exchangeDeclare']['autoDelete']	说明：交换机声明，自动删除。[true][默认]
	 * ················· [可空/多选_A/必须](unknown_type)	$params['AAAAA']	说明：AAAAA
	 * @return $this
	 */
	public function exchangeDeclareToPools( string $channelSid = 'default', array $params = [] ):self {
		// 创建
		$channel = $this->rabbitMQ->newChannel();
		$exchange = $channel->newExchange( $this->getChannelSidPrefix() . $channelSid );
		$exchange->exchangeDeclare(
			AMQPExchangeType::TOPIC,
			$params['exchangeDeclare']['passive'] ?? false,
			$params['exchangeDeclare']['durable'] ?? false,
			$params['exchangeDeclare']['autoDelete'] ?? true,
		);
		// $exchange->getChannel()->confirm_select();
		//
		$this->exchangePool[ $channelSid ] = $exchange;
		//
		return $this;
	}

	/**
	 * @param string $channelSid
	 * @param string $roomSid
	 * @params array $params 参数数组
	 * ················· [可空](bool)	$params['queueDeclare']['passive']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['durable']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['exclusive']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['autoDelete']	说明：队列声明，自动删除。[true][默认]
	 * ················· [可空/多选_A/必须](unknown_type)	$params['AAAAA']	说明：AAAAA
	 * @return com\Lib\Com\RabbitMQ\_Channel\_Queue
	 */
	public function getQueueOfPools( string $channelSid, string $roomSid, array $params = [] ):com\Lib\Com\RabbitMQ\_Channel\_Queue {
		//
		$queueName = $this->buildQueueName( $channelSid, $roomSid );
		//
		if( ! array_key_exists( $queueName, $this->queuePools ) ){
			$this->queueDeclareToPools( $channelSid, $roomSid, $params );
		}
		//
		return $this->queuePools[ $queueName ];
	}

	/**
	 * 声明队列（会覆盖pool）
	 * @param string $roomSid
	 * @params array $params 参数数组
	 * ················· [可空](bool)	$params['queueDeclare']['passive']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['durable']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['exclusive']	说明：队列声明。[false][默认]
	 * ················· [可空](bool)	$params['queueDeclare']['autoDelete']	说明：队列声明，自动删除。[true][默认]
	 * ················· [可空/多选_A/必须](unknown_type)	$params['AAAAA']	说明：AAAAA
	 * @return $this
	 */
	public function queueDeclareToPools( string $channelSid, string $roomSid, array $params = [] ):self {
		$exchangeNameAsPrefix = $this->getChannelSidPrefix() . $channelSid;
		//
		$routeKey = "room.msg.{$roomSid}";
		$queueName = $this->buildQueueName( $channelSid, $roomSid );
		// 初始化交换机
		$this->exchangeDeclareToPools( $channelSid );
		// 接收消息
		$queue = $this->rabbitMQ->newChannel()->newQueue( $queueName );
		$queue->declareQueue(
			$params['queueDeclare']['passive'] ?? false,
			$params['queueDeclare']['durable'] ?? false,
			$params['queueDeclare']['exclusive'] ?? false,
			$params['queueDeclare']['autoDelete'] ?? true
		)->bind( $exchangeNameAsPrefix, $routeKey );
		//
		$this->queuePools[ $queueName ] = $queue;
		//
		return $this;
	}


	/**
	 * 关闭池中的队列
	 * @param string $channelSid
	 * @param string $roomSid
	 * @return void
	 */
	public function closeQueueOfPools( string $channelSid, string $roomSid ) {
		$queueName = $this->buildQueueName( $channelSid, $roomSid );
		//
		if( array_key_exists( $queueName, $this->queuePools ) ){
			// 停止接收
			$this->queuePools[ $queueName ]->goConsume()->stop();
			// 关闭通道
			$this->queuePools[ $queueName ]->getChannel()->close();
			//
			unset( $this->queuePools[ $queueName ] );
		}
	}

	/**
	 * 构建队列名
	 * @param string $channelSid
	 * @param string $roomSid
	 * @return string
	 */
	private function buildQueueName( string $channelSid, string $roomSid ):string {
		$exchangeNameAsPrefix = $this->getChannelSidPrefix() . $channelSid;
		//
		$routeKey = "room.msg.{$roomSid}";
		return "{$exchangeNameAsPrefix}=>{$routeKey}";
	}

	/**
	 * @param \Closure $fn
	 * @return mixed
	 */
	private function callRabbitMQ( \Closure $fn ) {
		try{
			return call_user_func( $fn );
		} catch( \PhpAmqpLib\Exception\AMQPChannelClosedException | \PhpAmqpLib\Exception\AMQPRuntimeException $e ){
			echo com\Lib\Err\Exception::getSimpleShowExceptionInfo( $e );
			//
			$this->initRabbitMQ();
			//
			return $this->callRabbitMQ( $fn );
		} catch( \Throwable $e ){
			echo com\Lib\Err\Exception::getSimpleShowExceptionInfo( $e );
		}
	}


	/**
	 * @param string $channelSid
	 * @param string $roomSid
	 * @param $body
	 * @param array $headers 头信息
	 */
	public function send( string $channelSid, string $roomSid, $body, array $headers = [] ):void {
		//
		$this->callRabbitMQ( function() use ( $channelSid, $roomSid, $body, $headers ){
			$data = [
				'body' => $body,
			];
			// 加入头
			if( ! empty( $headers ) ){
				$data['headers'] = $headers;
			}
			// return $channel->newExchange( $this->getChannelSidPrefix() . $channelSid )->publish( json_encode( $data, JSON_UNESCAPED_UNICODE ), "room.msg.{$roomSid}" );
			// 声明房间
			$this->getQueueOfPools( $channelSid, $roomSid );
			// 发送给交换机
			return $this->getExchangeOfPools( $channelSid )->publish( json_encode( $data, JSON_UNESCAPED_UNICODE ), "room.msg.{$roomSid}" );
		} );
	}

	/**
	 * @param string $mailId
	 * @param $body
	 * @param array $headers 头信息
	 */
	public function sendMail( string $mailId, $body, array $headers = [] ):void {
		$mailStruct = explode( '@', $mailId );
		//
		$headers['toMailId'] = $mailId;
		//
		$this->send( $mailStruct[2] ?? '', $mailStruct[1], $body, $headers );
	}

	/**
	 * 消息处理器
	 * @param string $channelSid
	 * @param \PhpAmqpLib\Message\AMQPMessage $msg
	 */
	private function consumeHandle( string $channelSid, \PhpAmqpLib\Message\AMQPMessage $AMQPMsg ):void {
		$roomSid = substr( $AMQPMsg->getRoutingKey(), 9 );
		// 执行
		$this->receivePools->callReceiveOfRoomSid( $channelSid, $roomSid, $AMQPMsg );
	}

	/**
	 * 获取接收通道类
	 * @param int $receiveId
	 * @return thisServer\ReceivePools\Item|null
	 */
	public function getReceiveItem( int $receiveId ):thisServer\ReceivePools\Item|null {
		return $this->receivePools->getReceiveItem( $receiveId );
	}

	/**
	 * 增加接收处理器
	 * @param string $channelSid
	 * @param string $roomSid
	 * @param \Closure $callback
	 * @param int|null $receiveTimeoutS 接收超时（单位：秒） [null]不超时
	 * @return int 返回接收id
	 */
	public function receive( string $channelSid, string $roomSid, \Closure $callback, ?int $receiveTimeoutS = null ):int {
		// 初始化交换机
		$queue = $this->getQueueOfPools( $channelSid, $roomSid );
		// 运行接收器
		if( ! $queue->goConsume()->isRunStatus() ){
			// $queue->goConsume()->stop();
			//
			$queue->goConsume()->callback( function ( \PhpAmqpLib\Message\AMQPMessage $AMQPMsg ) use ( $channelSid, $queue ) {
				$this->consumeHandle( $channelSid, $AMQPMsg );
				//
				if( $queue->getChannel()->is_open() ){
					d('ack');
					$AMQPMsg->ack();
				}
			} )->runOfCoroutine();
		}
		//
		return $this->receivePools->add( $channelSid, $roomSid, $callback, function() use ( $queue, $roomSid, $channelSid ){
			// 删除pool
			$this->closeQueueOfPools( $channelSid, $roomSid );
		}, $receiveTimeoutS );
	}

	/**
	 * 等待接收处理器
	 * @param int $receiveId 接收id
	 * @param int|null $waitTimeoutS 等待超时时间。单位：秒
	 * @return bool 是否已结束 [true]正常结束 [fasle]超时或已关闭
	 */
	public function waitReceive( int $receiveId, ?int $waitTimeoutS = null ):bool {
		return $this->receivePools->waitReceive( $receiveId, $waitTimeoutS );
	}

	/**
	 * 是否存在接收处理器
	 * @param int $receiveId 接收id
	 * @return bool
	 */
	public function hasReceive( int $receiveId ):bool {
		return $this->receivePools->has( $receiveId );
	}

	/**
	 * 移除接收处理器
	 * @param int $receiveId 接收id
	 * @return bool
	 */
	public function stopReceive( int $receiveId ):bool {
		return $this->receivePools->rm( $receiveId );
	}

	/**
	 * 一般不需要手动关闭，连接需一直存在整个进程的周期
	 * @return self
	 */
	public function close(  ):self {
		if( $this->rabbitMQ !== null ){
			$this->rabbitMQ->close();
		}
		//
		return $this;
	}

	// ===============================
	//
	//
	// 下面为静态方法
	//
	//
	// ===============================

	/**
	 * @return $this
	 */
	public static function makeOfNew():self {
	    return self::make();
	}

}
